package com.navinfo.opentsp.platform.computing.serializer;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Charsets;
import com.navinfo.opentsp.platform.computing.dto.FaultInfoDto;
import com.navinfo.opentsp.platform.computing.dto.FaultItem;
import com.navinfo.opentsp.platform.computing.util.DateUtils;
import com.navinfo.opentsp.platform.location.protocol.common.LCChassisDataUpload;
import com.navinfo.opentsp.platform.location.protocol.common.LCFaultInfo;
import com.navinfo.opentsp.platform.location.protocol.common.LCFaultTimerUpload;
import com.navinfo.opentsp.platform.location.protocol.common.LCFaultUpload;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.serialization.EventSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;

/**
* ChassisUpload信息 A204 协议数据上传，
* HDFS_Sink的DBCUpload序列化器，拼接成字段间通过\001分割的字符串
*/
public class HDFSFaultTimerUploadSerializer implements EventSerializer {
    private static final Logger logger = LoggerFactory.getLogger(HDFSFaultTimerUploadSerializer.class);

    private final String APPEND_NEWLINE = "appendNewline";
    private final boolean appendNewline;
    private final String splitStr = "\001";
    private final OutputStream out;

    private HDFSFaultTimerUploadSerializer(OutputStream out, Context ctx) {
        this.appendNewline = ctx.getBoolean(APPEND_NEWLINE, true);
        this.out = out;
    }

    @Override
    public boolean supportsReopen() {
        return false;
    }
    @Override
    public void afterCreate() {}
    @Override
    public void afterReopen() {}
    @Override
    public void beforeClose() {}
    @Override
    public void flush() throws IOException {
        this.out.flush();
    }
    @Override
    public void write(Event event) throws IOException {
        try {
//            logger.error(event.toString());
            //拼写落盘到HDFS的字符串。字段间分隔符是\001，null通过\N表示
            LCFaultTimerUpload.FaultTimerUpload faultTimerUpload = LCFaultTimerUpload.FaultTimerUpload.parseFrom(event.getBody());
            StringBuilder faultTimerUploadStr = new StringBuilder(5000);

            faultTimerUploadStr.append(faultTimerUpload.getLatitude()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getLongitude()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getOriginalLat()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getOriginalLng()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getHeight()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getDirection()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getGpsDate()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getTotalDriveMileage()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getGpsSpeed()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getFaultCnt()).append(splitStr);
            faultTimerUploadStr.append(JSON.toJSON(getFaultInfoList(faultTimerUpload.getFaultListList()))).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getSourceAddress()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getCollectDate()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getDataLength()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getAtmosphericPressure()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getAtmosphericTemperature()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getFakeThrottleOpening()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getT50Status()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getThrottleOpening()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getDashboardSpeed()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getTotalMileage()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getSubTotalMileage()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getIntakeAirTemperature()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getExhaustGasTemperature()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getNoxInContr()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getDpfPressureDiff()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getNoxOutContr()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getUreaLevel()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getUreaTemp()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getReferTorque()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getEgrStatus()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getActualOutputTorque()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getEngineSpeed()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getOutFlow()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getInNox()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getOutNox()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getInFlow()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getWaterTemperature()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getCurrentGear()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getIntakeManifoldBoostPressure()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getIntakeManifoldBoostAirTemperature()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getOuttakeAirTemperature()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getRealTimeOilConsumption()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getAverageFuelConsumption()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getThrottle1Valve()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getEecuGear()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getOilPressure()).append(splitStr);
            faultTimerUploadStr.append(faultTimerUpload.getTerminalId()).append(splitStr);


            //添加字段只能在分割线之上添加,注意时间分区这块，终端传的时候是秒还是毫秒
            //------------------------------------------分割线----------------------------------------
            faultTimerUploadStr.append(DateUtils.format(faultTimerUpload.getGpsDate() * 1000, "yyyyMMdd"));
//            logger.error(dataJson);
            //将数据拼接后写入hdfs
            this.out.write(faultTimerUploadStr.toString().getBytes(Charsets.UTF_8));
            if (this.appendNewline) {
                this.out.write(10);
            }
        }catch (Exception e){
            logger.error("执行HDFS的HDFSFaultTimerUploadSerializer拦截器失败：" + e.getMessage(), e);
        }
    }

//    public static List<String> getMaintenanceEvents(List<LCLocationData.StaytimeParkingAddition> staytimeParkings){
//        List<String> staytimeParkingDtos = Lists.newArrayList();
//        if(staytimeParkings == null){
//            return staytimeParkingDtos;
//        }
//        for(LCLocationData.StaytimeParkingAddition staytimeParking : staytimeParkings){
//            staytimeParkingDtos.add(StaytimeParkingAdditionDto.parseString(staytimeParking));
//        }
//        return staytimeParkingDtos;
//    }


    public static class Builder implements EventSerializer.Builder {
        public Builder() { }
        @Override
        public EventSerializer build(Context context, OutputStream out) {
            return  new HDFSFaultTimerUploadSerializer(out, context);
        }
    }

    public String ListToString(List<Long> list){
        String listStr = "";
        for (Iterator<Long> iterator = list.iterator(); iterator.hasNext(); ) {
            Long next = iterator.next();
            listStr = listStr.concat("_");
            listStr = listStr.concat(String.valueOf(next));
        }

        return listStr.substring(1);

    }
    public List<FaultItem> getFaultInfoList(List<LCFaultTimerUpload.FaultItem> faultListData) {
        if (faultListData == null) {
            return Collections.emptyList();
        }
        List<FaultItem> faultItems = new ArrayList<>();
        for (LCFaultTimerUpload.FaultItem faultInfo : faultListData) {
            faultItems.add(FaultItem.protoToBean(faultInfo));
        }
        return faultItems;
    }

}
